using Microsoft.Extensions.Logging;
using SeataNet.Core.Configuration;
using SeataNet.Core.Exceptions;
using SeataNet.Core.Protocol;
using SeataNet.Core.Common;
using System.Collections.Concurrent;
using System.Net;
using System.Text.Json;

namespace SeataNet.Core.RPC
{
    /// <summary>
    /// RPC网络传输层
    /// </summary>
    public interface IRpcTransport
    {
        Task<T> SendAsync<T>(AbstractMessage message, string targetAddress, CancellationToken cancellationToken = default) where T : AbstractMessage;
        Task SendOnewayAsync(AbstractMessage message, string targetAddress, CancellationToken cancellationToken = default);
        void RegisterMessageHandler<T>(Func<T, Task<AbstractMessage>> handler) where T : AbstractMessage;
        void UnregisterMessageHandler<T>() where T : AbstractMessage;
        Task StartAsync(CancellationToken cancellationToken = default);
        Task StopAsync(CancellationToken cancellationToken = default);
        bool IsRunning { get; }
    }

    /// <summary>
    /// HTTP RPC网络传输层
    /// </summary>
    public class HttpRpcTransport : IRpcTransport, IDisposable
    {
        private readonly SeataConfiguration _configuration;
        private readonly ILogger<HttpRpcTransport> _logger;
        private readonly ILoggerFactory _loggerFactory;
        private readonly ConcurrentDictionary<Type, Func<AbstractMessage, Task<AbstractMessage>>> _messageHandlers;
        private readonly IMessageSerializer _serializer;
        private readonly HttpClient _httpClient;
        private readonly HttpListener _httpListener;
        private readonly CancellationTokenSource _cancellationTokenSource;
        private readonly Task _processingTask;
        private volatile bool _isRunning = false;
        private volatile bool _disposed = false;

        public bool IsRunning => _isRunning;

        public HttpRpcTransport(
            SeataConfiguration configuration,
            ILogger<HttpRpcTransport> logger,
            ILoggerFactory loggerFactory,
            IMessageSerializer? serializer = null,
            HttpClient? httpClient = null)
        {
            _configuration = configuration;
            _logger = logger;
            _loggerFactory = loggerFactory;
            _messageHandlers = new ConcurrentDictionary<Type, Func<AbstractMessage, Task<AbstractMessage>>>();
            _serializer = serializer ?? new JsonMessageSerializer(_loggerFactory.CreateLogger<JsonMessageSerializer>());
            _httpClient = httpClient ?? CreateHttpClient();
            _cancellationTokenSource = new CancellationTokenSource();
            
            // 解析监听地址
            var listenAddress = ParseListenAddress(configuration.Transport.ListenAddress);
            _httpListener = new HttpListener();
            _httpListener.Prefixes.Add(listenAddress);
            
            // 启动处理任务
            _processingTask = Task.Run(ProcessRequestsAsync, _cancellationTokenSource.Token);
        }

        public async Task<T> SendAsync<T>(AbstractMessage message, string targetAddress, CancellationToken cancellationToken = default) where T : AbstractMessage
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(nameof(HttpRpcTransport));
            }

            if (message == null)
            {
                throw new ArgumentNullException(nameof(message));
            }

            if (string.IsNullOrWhiteSpace(targetAddress))
            {
                throw new ArgumentNullException(nameof(targetAddress));
            }

            var messageType = message.GetType();
            _logger.LogDebug($"发送消息: {messageType.Name}, ID: {message.MessageId}, 目标: {targetAddress}");

            try
            {
                var serializedData = _serializer.Serialize(message);
                var content = new ByteArrayContent(serializedData);
                
                // 添加请求头
                content.Headers.Add("Content-Type", "application/octet-stream");
                content.Headers.Add("X-Message-Type", message.MessageCode.ToString());
                
                var response = await _httpClient.PostAsync(
                    $"{targetAddress}/seata/message",
                    content,
                    cancellationToken);
                
                if (!response.IsSuccessStatusCode)
                {
                    throw new RpcException($"服务器返回错误状态码: {response.StatusCode}");
                }
                
                var responseData = await response.Content.ReadAsByteArrayAsync(cancellationToken);
                var responseMessage = _serializer.Deserialize(responseData);
                
                if (!(responseMessage is T typedResponse))
                {
                    throw new RpcException($"响应消息类型不匹配: 期望 {typeof(T).Name}, 实际 {responseMessage?.GetType().Name}");
                }
                
                _logger.LogDebug($"接收响应: {typedResponse.GetType().Name}, ID: {typedResponse.MessageId}");
                return typedResponse;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, $"发送消息失败: {messageType.Name}, 目标: {targetAddress}");
                throw new RpcException($"发送消息失败: {messageType.Name}", ex);
            }
        }

        public async Task SendOnewayAsync(AbstractMessage message, string targetAddress, CancellationToken cancellationToken = default)
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(nameof(HttpRpcTransport));
            }

            if (message == null)
            {
                throw new ArgumentNullException(nameof(message));
            }

            if (string.IsNullOrWhiteSpace(targetAddress))
            {
                throw new ArgumentNullException(nameof(targetAddress));
            }

            var messageType = message.GetType();
            _logger.LogDebug($"发送单向消息: {messageType.Name}, ID: {message.MessageId}, 目标: {targetAddress}");

            try
            {
                var serializedData = _serializer.Serialize(message);
                var content = new ByteArrayContent(serializedData);
                
                // 添加请求头
                content.Headers.Add("Content-Type", "application/octet-stream");
                content.Headers.Add("X-Message-Type", message.MessageCode.ToString());
                content.Headers.Add("X-Oneway", "true");
                
                var response = await _httpClient.PostAsync(
                    $"{targetAddress}/seata/message",
                    content,
                    cancellationToken);
                
                // 单向消息不关心响应内容，只关心是否成功发送
                if (!response.IsSuccessStatusCode)
                {
                    throw new RpcException($"服务器返回错误状态码: {response.StatusCode}");
                }
                
                _logger.LogDebug($"单向消息发送成功: {messageType.Name}, ID: {message.MessageId}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, $"发送单向消息失败: {messageType.Name}, 目标: {targetAddress}");
                throw new RpcException($"发送单向消息失败: {messageType.Name}", ex);
            }
        }

        public void RegisterMessageHandler<T>(Func<T, Task<AbstractMessage>> handler) where T : AbstractMessage
        {
            if (handler == null)
            {
                throw new ArgumentNullException(nameof(handler));
            }

            var messageType = typeof(T);
            _messageHandlers[messageType] = async (message) =>
            {
                if (message is T typedMessage)
                {
                    return await handler(typedMessage);
                }
                
                throw new InvalidOperationException($"消息类型不匹配: 期望 {messageType.Name}, 实际 {message.GetType().Name}");
            };

            _logger.LogDebug($"注册消息处理器: {messageType.Name}");
        }

        public void UnregisterMessageHandler<T>() where T : AbstractMessage
        {
            var messageType = typeof(T);
            _messageHandlers.TryRemove(messageType, out _);
            _logger.LogDebug($"取消注册消息处理器: {messageType.Name}");
        }

        public async Task StartAsync(CancellationToken cancellationToken = default)
        {
            if (_disposed)
            {
                throw new ObjectDisposedException(nameof(HttpRpcTransport));
            }

            if (_isRunning)
            {
                return;
            }

            try
            {
                _httpListener.Start();
                _isRunning = true;
                _logger.LogInformation($"HTTP RPC传输层已启动，监听地址: {_configuration.Transport.ListenAddress}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "启动HTTP RPC传输层失败");
                throw new RpcException("启动HTTP RPC传输层失败", ex);
            }
        }

        public async Task StopAsync(CancellationToken cancellationToken = default)
        {
            if (!_isRunning)
            {
                return;
            }

            try
            {
                _httpListener.Stop();
                _isRunning = false;
                _logger.LogInformation("HTTP RPC传输层已停止");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "停止HTTP RPC传输层失败");
                throw new RpcException("停止HTTP RPC传输层失败", ex);
            }
        }

        private async Task ProcessRequestsAsync()
        {
            try
            {
                while (!_cancellationTokenSource.Token.IsCancellationRequested)
                {
                    try
                    {
                        var context = await _httpListener.GetContextAsync();
                        _ = Task.Run(async () => await ProcessRequestAsync(context), _cancellationTokenSource.Token);
                    }
                    catch (ObjectDisposedException)
                    {
                        // HttpListener被释放，退出循环
                        break;
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "处理请求时发生错误");
                    }
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理请求任务异常退出");
            }
        }

        private async Task ProcessRequestAsync(HttpListenerContext context)
        {
            try
            {
                var request = context.Request;
                var response = context.Response;
                
                // 检查请求方法
                if (request.HttpMethod != "POST")
                {
                    response.StatusCode = 405; // Method Not Allowed
                    response.Close();
                    return;
                }
                
                // 检查是否是心跳请求
                if (request.Url?.AbsolutePath.EndsWith("/heartbeat") == true)
                {
                    await ProcessHeartbeatRequestAsync(context);
                    return;
                }
                
                // 检查是否是消息请求
                if (request.Url?.AbsolutePath.EndsWith("/message") == true)
                {
                    await ProcessMessageRequestAsync(context);
                    return;
                }
                
                // 未知请求
                response.StatusCode = 404; // Not Found
                response.Close();
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理HTTP请求时发生错误");
                
                try
                {
                    var response = context.Response;
                    response.StatusCode = 500; // Internal Server Error
                    response.Close();
                }
                catch
                {
                    // 忽略关闭响应时的错误
                }
            }
        }

        private async Task ProcessHeartbeatRequestAsync(HttpListenerContext context)
        {
            var request = context.Request;
            var response = context.Response;
            
            try
            {
                // 读取请求数据
                var requestData = await ReadRequestDataAsync(request);
                var heartbeatRequest = (HeartbeatRequest)_serializer.Deserialize(requestData);
                
                // 处理心跳请求
                var heartbeatResponse = new HeartbeatResponse
                {
                    MessageId = heartbeatRequest.MessageId,
                    ResultCode = ResultCode.Success
                };
                
                // 发送响应
                var responseData = _serializer.Serialize(heartbeatResponse);
                await WriteResponseDataAsync(response, responseData);
                
                _logger.LogDebug("处理心跳请求成功");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理心跳请求失败");
                
                response.StatusCode = 500; // Internal Server Error
                response.Close();
            }
        }

        private async Task ProcessMessageRequestAsync(HttpListenerContext context)
        {
            var request = context.Request;
            var response = context.Response;
            
            try
            {
                // 读取请求数据
                var requestData = await ReadRequestDataAsync(request);
                var message = _serializer.Deserialize(requestData);
                
                // 检查是否是单向消息
                var isOneway = request.Headers["X-Oneway"] == "true";
                
                if (isOneway)
                {
                    // 单向消息，不需要响应
                    await ProcessMessageAsync(message);
                    response.StatusCode = 200; // OK
                    response.Close();
                    return;
                }
                
                // 双向消息，需要响应
                var responseMessage = await ProcessMessageAsync(message);
                
                // 发送响应
                var responseData = _serializer.Serialize(responseMessage);
                await WriteResponseDataAsync(response, responseData);
                
                _logger.LogDebug($"处理消息请求成功: {message.GetType().Name}");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理消息请求失败");
                
                response.StatusCode = 500; // Internal Server Error
                response.Close();
            }
        }

        private async Task<AbstractMessage> ProcessMessageAsync(AbstractMessage message)
        {
            var messageType = message.GetType();
            
            if (!_messageHandlers.TryGetValue(messageType, out var handler))
            {
                throw new RpcException($"未找到消息处理器: {messageType.Name}");
            }
            
            return await handler(message);
        }

        private async Task<byte[]> ReadRequestDataAsync(HttpListenerRequest request)
        {
            using var memoryStream = new MemoryStream();
            await request.InputStream.CopyToAsync(memoryStream);
            return memoryStream.ToArray();
        }

        private async Task WriteResponseDataAsync(HttpListenerResponse response, byte[] data)
        {
            response.ContentType = "application/octet-stream";
            response.ContentLength64 = data.Length;
            
            await response.OutputStream.WriteAsync(data, 0, data.Length);
            response.Close();
        }

        private HttpClient CreateHttpClient()
        {
            var httpClient = new HttpClient();
            httpClient.Timeout = _configuration.Transport.Timeout;
            return httpClient;
        }

        private string ParseListenAddress(string listenAddress)
        {
            if (string.IsNullOrWhiteSpace(listenAddress))
            {
                return "http://localhost:8091/";
            }
            
            var address = listenAddress.Trim();
            
            // 如果没有协议前缀，添加http://
            if (!address.StartsWith("http://") && !address.StartsWith("https://"))
            {
                address = "http://" + address;
            }
            
            // 如果没有端口，添加默认端口8091
            if (!address.Contains(":") || address.EndsWith(":"))
            {
                address = address.TrimEnd(':') + ":8091";
            }
            
            // 确保以/结尾
            if (!address.EndsWith("/"))
            {
                address += "/";
            }
            
            return address;
        }

        public void Dispose()
        {
            if (_disposed)
            {
                return;
            }

            _disposed = true;
            
            try
            {
                StopAsync().GetAwaiter().GetResult();
            }
            catch
            {
                // 忽略停止时的错误
            }

            _cancellationTokenSource?.Cancel();
            _cancellationTokenSource?.Dispose();

            _httpListener?.Stop();
            _httpListener?.Close();

            _httpClient?.Dispose();

            _messageHandlers.Clear();
        }
    }
}